package com.shujia.base;

import java.io.*;
import java.util.HashMap;

public class MapTask implements Runnable{
    private File block;
    private String blockName;
    BufferedReader br = null;
    BufferedWriter bw = null;

    public MapTask(File block,String blockName){
        this.block = block;
        this.blockName = blockName;
        try {
            br = new BufferedReader(new FileReader(block.getAbsolutePath()));
            //blockName - "block0"
            //part-m-00000
            bw = new BufferedWriter(new FileWriter("hadoop/data/maps/part-m-0000"+blockName.substring(5)));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        try {
            //编写每个map任务要做的事情

            //创建Map集合
            HashMap<String, Long> map = new HashMap<>();

            String line = null;
            while ((line = br.readLine()) != null) {
                //1500100003,单乐蕊,22,女,理科六班
                String clazz = line.split(",")[4];
                //判断集合中是否存在该班级为键
                if (map.containsKey(clazz)) {
                    //将原来的值+1， 重新赋值回去
                    map.put(clazz, map.get(clazz) + 1L);
                } else {
                    //说明是第一次出现，以班级作为键，1L作为值存储
                    map.put(clazz, 1L);
                }
            }

            //遍历map集合得到每个班级的人数
            map.forEach((k, v) -> {
                try {
                    bw.write(k+" "+v);
                    bw.newLine();
                    bw.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(bw!=null){
                try {
                    bw.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            if(br!=null){
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}